package com.xiehua.task.actor;

import akka.actor.ActorSelection;
import akka.pattern.Patterns;
import akka.persistence.AbstractPersistentActorWithAtLeastOnceDelivery;
import com.xiehua.task.actor.cmd.Cmd;
import com.xiehua.task.job.bucket.ReadyBucket;
import com.xiehua.task.job.pool.Job;
import com.xiehua.task.redis.RedisTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

public class TaskDisposePersistentActor extends AbstractPersistentActorWithAtLeastOnceDelivery {

    private static final Logger LOGGER = LoggerFactory.getLogger(TaskDisposePersistentActor.class);

//    private List<Job> readyJob;

    @Override
    public String persistenceId() {
        return "xiehua-task-id";
    }

    @Override
    public Receive createReceiveRecover() {
        return null;
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder().match(Cmd.class, t -> t.getCmd().equals(Cmd.EVT_TASK_DISPOSE), s -> {
            if (LOGGER.isDebugEnabled()) LOGGER.debug("收到消息type：{}", s.getCmd());
            ReadyBucket readyBucket = new ReadyBucket(RedisTemplate.getJedis());
            List<Job> readyJob = readyBucket.pollTaskBatch();
            readyBucket.clear();
            LOGGER.info("通知客户端处理消息条数:{}", readyJob.size());
            // 将消息发给下一个处理者TaskNotifyActor
            //ActorSelection taskNotifyActor = getContext().actorSelection("akka://xiehua_task_server_actors_sys/user/TaskDisposeActor/");
            //taskNotifyActor.tell(cmd,self());
            //dispatcher(readyJob, taskNotifyActor);
        }).match(Cmd.class, t -> t.getCmd().equals(Cmd.CMD_TASK_NOTIFY_ASYNC_RESP), s -> {
            LOGGER.info("JOB已通知客户端:{}", s.getT());
            //deleteSnapshot();
        }).matchAny(t -> {
            LOGGER.error("没有对应Actor处理本次消息~");
            unhandled(t);
        }).build();

    }

    /**
     * 任务快照分发~
     **/
    private void snapshot(List<Job> readyJob, ActorSelection taskNotifyActor) {
        saveSnapshot(readyJobCopy(readyJob));
        persistAll(readyJob, t -> {//持久化
            Cmd cmd = new Cmd(Cmd.CMD_TASK_NOTIFY, t);
            Future<Object> future = Patterns.ask(taskNotifyActor, cmd, 2000);
            Patterns.pipe(future, getContext().getSystem().dispatcher()).to(getSelf());
        });
    }

    /**
     * 任务分发~
     **/
    private void dispatcher(List<Job> readyJob, ActorSelection taskNotifyActor) {
        readyJob.forEach(s -> {
            Cmd cmd = new Cmd(Cmd.CMD_TASK_NOTIFY, s);
            Future<Object> future = Patterns.ask(taskNotifyActor, cmd, 2000);
            Patterns.pipe(future, getContext().getSystem().dispatcher()).to(getSelf());
        });
    }

    /***
     * job list copy
     * **/
    private ArrayList<Job> readyJobCopy(List<Job> list) {
        ArrayList<Job> copy = new ArrayList(Arrays.asList(new Job[list.size()]));
        Collections.copy(copy, list);
        return copy;
    }

    public void onSuccess() throws Exception {

        //等等future返回
//        Future<Object> future = Patterns.ask(this, 5, 1000);
//        int result = (int) Await.result(future, Duration.create(3, TimeUnit.SECONDS));
//        System.out.println("result:" + result);

        //不等待返回值，直接重定向到其他actor，有返回值来的时候将会重定向到printActor

//        Future<Object> future1 = Patterns.ask(workerActor, 8, 1000);
//        Patterns.pipe(future1, system.dispatcher()).to(printActor);
    }


}
